/*
 *  Copyright 2012 The WebRTC Project Authors. All rights reserved.
 *
 *  Use of this source code is governed by a BSD-style license
 *  that can be found in the LICENSE file in the root of the source
 *  tree. An additional intellectual property rights grant can be found
 *  in the file PATENTS.  All contributing project authors may
 *  be found in the AUTHORS file in the root of the source tree.
 */

#include "conductor.h"

#include <stddef.h>
#include <stdint.h>
#include <thread>
#include <memory>
#include <utility>
#include <vector>

#include "third_party/abseil-cpp/absl/memory/memory.h"
#include "third_party/abseil-cpp/absl/types/optional.h"
#include "api/audio/audio_mixer.h"
#include "api/audio_codecs/audio_decoder_factory.h"
#include "api/audio_codecs/audio_encoder_factory.h"
#include "api/audio_codecs/builtin_audio_decoder_factory.h"
#include "api/audio_codecs/builtin_audio_encoder_factory.h"
#include "api/audio_options.h"
#include "api/create_peerconnection_factory.h"
#include "api/rtp_sender_interface.h"
#include "api/video_codecs/builtin_video_decoder_factory.h"
#include "api/video_codecs/builtin_video_encoder_factory.h"
#include "api/video_codecs/video_decoder_factory.h"
#include "api/video_codecs/video_encoder_factory.h"
#include "defaults.h"
#include "modules/audio_device/include/audio_device.h"
#include "modules/audio_processing/include/audio_processing.h"
#include "modules/video_capture/video_capture.h"
#include "modules/video_capture/video_capture_factory.h"
#include "p2p/base/port_allocator.h"
#include "pc/video_track_source.h"
#include "rtc_base/checks.h"
#include "rtc_base/logging.h"
#include "rtc_base/ref_counted_object.h"
#include "rtc_base/rtc_certificate_generator.h"
#include "capture/vcm_capture.h"
#include "capture/external_video_capturer.h"
#include "json.hpp"
using json = nlohmann::json;

namespace {
// Names used for a IceCandidate JSON object.
    const char kCandidateSdpMidName[] = "sdpMid";
    const char kCandidateSdpMlineIndexName[] = "sdpMLineIndex";
    const char kCandidateSdpName[] = "candidate";

// Names used for a SessionDescription JSON object.
    const char kSessionDescriptionTypeName[] = "type";
    const char kSessionDescriptionSdpName[] = "sdp";

    class DummySetSessionDescriptionObserver
            : public webrtc::SetSessionDescriptionObserver {
    public:
        static DummySetSessionDescriptionObserver* Create() {
            return new rtc::RefCountedObject<DummySetSessionDescriptionObserver>();
        }
        virtual void OnSuccess() { RTC_LOG(INFO) << __FUNCTION__; }
        virtual void OnFailure(webrtc::RTCError error) {
            RTC_LOG(INFO) << __FUNCTION__ << " " << ToString(error.type()) << ": "
                          << error.message();
        }
    };

    class ExternalCapturerTrackSource : public webrtc::VideoTrackSource {
    public:
        static rtc::scoped_refptr<ExternalCapturerTrackSource> Create() {
            auto source = new rtc::RefCountedObject<ExternalCapturerTrackSource>
                    ( ExternalVideoCapturer::Create());
            source->StartCapture();
            return source;
        }

    protected:
        explicit ExternalCapturerTrackSource(
                std::unique_ptr<ExternalVideoCapturer> capturer)
                : VideoTrackSource(/*remote=*/false), capturer_(std::move(capturer)) {}

        ~ExternalCapturerTrackSource(){
            stop_ = true;
            if(thread_ && thread_->joinable())
                thread_->join();

        }

        void StartCapture(){
            thread_ = std::make_unique<std::thread>([this]{
                const size_t kWidth = 640;
                const size_t kHeight = 480;

                FILE* fp = NULL;
                fp = fopen("/Users/malei/Movies/anqi.yuv", "rb");
                if(!fp)
                    printf("failed to open yuv file.");
                if (!fp)return;

                unsigned int pixel_w = 1280;
                unsigned int pixel_h = 720;

                unsigned int bpp = 12;

                int len = pixel_w * pixel_h * bpp / 8;
                unsigned char* data = new unsigned char [len];


                static int count = 0;

                while (!stop_)
                {
                    if (fread(data, 1, pixel_w * pixel_h * bpp / 8, fp) != pixel_w * pixel_h * bpp / 8) {
                        // Loop
                        fseek(fp, 0, SEEK_SET);
                        fread(data, 1, pixel_w * pixel_h * bpp / 8, fp);

                    }

                    usleep(50*1000);
//                    struct timeval NTP_value;
//
//                    gettimeofday(&NTP_value, NULL);
//        int64_t t = GetNTPTimestamp();

//                    RTC_LOG(LS_VERBOSE) << "capturing video...";
                    capturer_->OnExternalVideoCaptured(data, len, pixel_w, pixel_h, 0);

                }

                delete [] data;
                if (fp) fclose(fp);

            });
        }

    private:
        rtc::VideoSourceInterface<webrtc::VideoFrame>* source() override {
            return capturer_.get();
        }
        std::unique_ptr<ExternalVideoCapturer> capturer_;
        std::unique_ptr<std::thread> thread_;
        bool stop_ {false};
    };

    class CapturerTrackSource : public webrtc::VideoTrackSource {
    public:
        static rtc::scoped_refptr<CapturerTrackSource> Create() {
            const size_t kWidth = 640;
            const size_t kHeight = 480;
            const size_t kFps = 30;
            std::unique_ptr<VcmCapturer> capturer;
            std::unique_ptr<webrtc::VideoCaptureModule::DeviceInfo> info(
                    webrtc::VideoCaptureFactory::CreateDeviceInfo());
            if (!info) {
                return nullptr;
            }
            int num_devices = info->NumberOfDevices();
            for (int i = 0; i < num_devices; ++i) {
                capturer = absl::WrapUnique(
                        VcmCapturer::Create(kWidth, kHeight, kFps, i));
                if (capturer) {
                    return new rtc::RefCountedObject<CapturerTrackSource>(
                            std::move(capturer));
                }
            }

            return nullptr;
        }

    protected:
        explicit CapturerTrackSource(
                std::unique_ptr<VcmCapturer> capturer)
                : VideoTrackSource(/*remote=*/false), capturer_(std::move(capturer)) {}

    private:
        rtc::VideoSourceInterface<webrtc::VideoFrame>* source() override {
            return capturer_.get();
        }
        std::unique_ptr<VcmCapturer> capturer_;
    };

}  // namespace

Conductor::Conductor(PeerConnectionClient* client, MainWindow* main_wnd)
        : peer_id_(-1), loopback_(false), client_(client), main_wnd_(main_wnd) {
    client_->RegisterObserver(this);
    main_wnd->RegisterObserver(this);
}

Conductor::~Conductor() {
    RTC_DCHECK(!peer_connection_);
}

bool Conductor::connection_active() const {
    return peer_connection_ != nullptr;
}

void Conductor::Close() {
    client_->SignOut();
    DeletePeerConnection();
}

bool Conductor::InitializePeerConnection() {
    RTC_DCHECK(!peer_connection_factory_);
    RTC_DCHECK(!peer_connection_);

    peer_connection_factory_ = webrtc::CreatePeerConnectionFactory(
            nullptr /* network_thread */, nullptr /* worker_thread */,
            nullptr /* signaling_thread */, nullptr /* default_adm */,
            webrtc::CreateBuiltinAudioEncoderFactory(),
            webrtc::CreateBuiltinAudioDecoderFactory(),
            webrtc::CreateBuiltinVideoEncoderFactory(),
            webrtc::CreateBuiltinVideoDecoderFactory(), nullptr /* audio_mixer */,
            nullptr /* audio_processing */);

    if (!peer_connection_factory_) {
        main_wnd_->MessageBox("Error", "Failed to initialize PeerConnectionFactory",
                              true);
        DeletePeerConnection();
        return false;
    }

    if (!CreatePeerConnection(/*dtls=*/true)) {
        main_wnd_->MessageBox("Error", "CreatePeerConnection failed", true);
        DeletePeerConnection();
    }

    AddTracks();

    return peer_connection_ != nullptr;
}

bool Conductor::ReinitializePeerConnectionForLoopback() {
    loopback_ = true;
    std::vector<rtc::scoped_refptr<webrtc::RtpSenderInterface>> senders =
            peer_connection_->GetSenders();
    peer_connection_ = nullptr;
    if (CreatePeerConnection(/*dtls=*/false)) {
        for (const auto& sender : senders) {
            peer_connection_->AddTrack(sender->track(), sender->stream_ids());
        }
        peer_connection_->CreateOffer(
                this, webrtc::PeerConnectionInterface::RTCOfferAnswerOptions());
    }
    return peer_connection_ != nullptr;
}

bool Conductor::CreatePeerConnection(bool dtls) {
    RTC_DCHECK(peer_connection_factory_);
    RTC_DCHECK(!peer_connection_);

    webrtc::PeerConnectionInterface::RTCConfiguration config;
    config.sdp_semantics = webrtc::SdpSemantics::kUnifiedPlan;
    config.enable_dtls_srtp = dtls;
    webrtc::PeerConnectionInterface::IceServer server;
    server.uri = GetPeerConnectionString();
    config.servers.push_back(server);

    peer_connection_ = peer_connection_factory_->CreatePeerConnection(
            config, nullptr, nullptr, this);
    return peer_connection_ != nullptr;
}

void Conductor::DeletePeerConnection() {
    main_wnd_->StopLocalRenderer();
    main_wnd_->StopRemoteRenderer();
    peer_connection_ = nullptr;
    peer_connection_factory_ = nullptr;
    peer_id_ = -1;
    loopback_ = false;
}

void Conductor::EnsureStreamingUI() {
    RTC_DCHECK(peer_connection_);
    if (main_wnd_->IsWindow()) {
        if (main_wnd_->current_ui() != MainWindow::STREAMING)
            main_wnd_->SwitchToStreamingUI();
    }
}

//
// PeerConnectionObserver implementation.
//

void Conductor::OnAddTrack(
        rtc::scoped_refptr<webrtc::RtpReceiverInterface> receiver,
        const std::vector<rtc::scoped_refptr<webrtc::MediaStreamInterface>>&
        streams) {
    RTC_LOG(INFO) << __FUNCTION__ << " [remote:] " << receiver->id();
    UIThreadCallback(NEW_TRACK_ADDED,
                                     receiver->track().release());
}

void Conductor::OnRemoveTrack(
        rtc::scoped_refptr<webrtc::RtpReceiverInterface> receiver) {
    RTC_LOG(INFO) << __FUNCTION__ << " [remote:] " << receiver->id();
    UIThreadCallback(TRACK_REMOVED, receiver->track().release());
}

void Conductor::OnIceCandidate(const webrtc::IceCandidateInterface* candidate) {
    RTC_LOG(INFO) << __FUNCTION__ << " " << candidate->sdp_mline_index();
    // For loopback test. To save some connecting delay.
    if (loopback_) {
        if (!peer_connection_->AddIceCandidate(candidate)) {
            RTC_LOG(WARNING) << "Failed to apply the received candidate";
        }
        return;
    }


    json jmessage;

    jmessage[kCandidateSdpMidName] = candidate->sdp_mid();
    jmessage[kCandidateSdpMlineIndexName] = candidate->sdp_mline_index();
    std::string sdp;
    if (!candidate->ToString(&sdp)) {
        RTC_LOG(LS_ERROR) << "Failed to serialize candidate";
        return;
    }
    jmessage[kCandidateSdpName] = sdp;
    auto msg = jmessage.dump(8);
    SendMessage(msg);
}

//
// PeerConnectionClientObserver implementation.
//

void Conductor::OnSignedIn() {
    RTC_LOG(INFO) << __FUNCTION__;
    main_wnd_->SwitchToPeerList(client_->peers());
}

void Conductor::OnDisconnected() {
    RTC_LOG(INFO) << __FUNCTION__;

    DeletePeerConnection();

    if (main_wnd_->IsWindow())
        main_wnd_->SwitchToConnectUI();
}

void Conductor::OnPeerConnected(int id, const std::string& name) {
    RTC_LOG(INFO) << __FUNCTION__ <<  " " << id << "-" << name;
    // Refresh the list if we're showing it.
    if (main_wnd_->current_ui() == MainWindow::LIST_PEERS)
        main_wnd_->SwitchToPeerList(client_->peers());
}

void Conductor::OnPeerDisconnected(int id) {
    RTC_LOG(INFO) << __FUNCTION__ << " " << id;
    if (id == peer_id_) {
        RTC_LOG(INFO) << "Our peer disconnected";
        UIThreadCallback(PEER_CONNECTION_CLOSED, NULL);
    } else {
        // Refresh the list if we're showing it.
        if (main_wnd_->current_ui() == MainWindow::LIST_PEERS)
            main_wnd_->SwitchToPeerList(client_->peers());
    }
}

void Conductor::OnMessageFromPeer(int peer_id, const std::string& message) {
    RTC_DCHECK(peer_id_ == peer_id || peer_id_ == -1);
    RTC_DCHECK(!message.empty());

    if (!peer_connection_.get()) {
        RTC_DCHECK(peer_id_ == -1);
        peer_id_ = peer_id;

        if (!InitializePeerConnection()) {
            RTC_LOG(LS_ERROR) << "Failed to initialize our PeerConnection instance";
            client_->SignOut();
            return;
        }
    } else if (peer_id != peer_id_) {
        RTC_DCHECK(peer_id_ != -1);
        RTC_LOG(WARNING)
        << "Received a message from unknown peer while already in a "
           "conversation with a different peer.";
        return;
    }

    json jmessage;
    try{
        jmessage = json::parse(message);
    }catch(...){
        RTC_LOG(WARNING) << "Received unknown message. " << message;
    }

    std::string type_str;
    std::string json_object;
    try{
        type_str = jmessage[kSessionDescriptionTypeName];
    }catch (...){
        RTC_LOG(LS_WARNING) << "parse type string failed";
    }


    if (!type_str.empty()) {
        if (type_str == "offer-loopback") {
            // This is a loopback call.
            // Recreate the peerconnection with DTLS disabled.
            if (!ReinitializePeerConnectionForLoopback()) {
                RTC_LOG(LS_ERROR) << "Failed to initialize our PeerConnection instance";
                DeletePeerConnection();
                client_->SignOut();
            }
            return;
        }
        absl::optional<webrtc::SdpType> type_maybe =
                webrtc::SdpTypeFromString(type_str);
        if (!type_maybe) {
            RTC_LOG(LS_ERROR) << "Unknown SDP type: " << type_str;
            return;
        }
        webrtc::SdpType type = *type_maybe;
        std::string sdp;
        if (!jmessage.contains(kSessionDescriptionSdpName)) {
            RTC_LOG(WARNING) << "Can't parse received session description message.";
            return;
        }
        sdp = jmessage[kSessionDescriptionSdpName];
        webrtc::SdpParseError error;
        std::unique_ptr<webrtc::SessionDescriptionInterface> session_description =
                webrtc::CreateSessionDescription(type, sdp, &error);
        if (!session_description) {
            RTC_LOG(WARNING) << "Can't parse received session description message. "
                                "SdpParseError was: "
                             << error.description;
            return;
        }
        RTC_LOG(INFO) << " Received session description :" << message;
        peer_connection_->SetRemoteDescription(
                DummySetSessionDescriptionObserver::Create(),
                session_description.release());
        if (type == webrtc::SdpType::kOffer) {
            peer_connection_->CreateAnswer(
                    this, webrtc::PeerConnectionInterface::RTCOfferAnswerOptions());
        }
    } else {
        std::string sdp_mid;
        int sdp_mlineindex = 0;
        std::string sdp;

        try {
            sdp_mid = jmessage[kCandidateSdpMidName];
            sdp_mlineindex = jmessage[kCandidateSdpMlineIndexName];
            sdp = jmessage[kCandidateSdpName];
        }catch (...){
            RTC_LOG(WARNING) << "Can't parse received message.";
            return;
        }

        webrtc::SdpParseError error;
        std::unique_ptr<webrtc::IceCandidateInterface> candidate(
                webrtc::CreateIceCandidate(sdp_mid, sdp_mlineindex, sdp, &error));
        if (!candidate.get()) {
            RTC_LOG(WARNING) << "Can't parse received candidate message. "
                                "SdpParseError was: "
                             << error.description;
            return;
        }
        if (!peer_connection_->AddIceCandidate(candidate.get())) {
            RTC_LOG(WARNING) << "Failed to apply the received candidate";
            return;
        }
        RTC_LOG(INFO) << " Received candidate :" << message;
    }
}

void Conductor::OnMessageSent(int err) {
    // Process the next pending message if any.
    UIThreadCallback(SEND_MESSAGE_TO_PEER, NULL);
}

void Conductor::OnServerConnectionFailure() {
    main_wnd_->MessageBox("Error", ("Failed to connect to " + server_).c_str(),
                          true);
}

//
// MainWndCallback implementation.
//

void Conductor::StartLogin(const std::string& server, int port) {
    if (client_->is_connected())
        return;
    server_ = server;
    client_->Connect(server, port, GetPeerName());
}

void Conductor::DisconnectFromServer() {
    if (client_->is_connected())
        client_->SignOut();
}

void Conductor::ConnectToPeer(int peer_id) {
    RTC_DCHECK(peer_id_ == -1);
    RTC_DCHECK(peer_id != -1);

    if (peer_connection_.get()) {
        main_wnd_->MessageBox(
                "Error", "We only support connecting to one peer at a time", true);
        return;
    }

    if (InitializePeerConnection()) {
        peer_id_ = peer_id;
        peer_connection_->CreateOffer(
                this, webrtc::PeerConnectionInterface::RTCOfferAnswerOptions());
    } else {
        main_wnd_->MessageBox("Error", "Failed to initialize PeerConnection", true);
    }
}

void Conductor::AddTracks() {
    if (!peer_connection_->GetSenders().empty()) {
        return;  // Already added tracks.
    }

    rtc::scoped_refptr<webrtc::AudioTrackInterface> audio_track(
            peer_connection_factory_->CreateAudioTrack(
                    kAudioLabel, peer_connection_factory_->CreateAudioSource(
                            cricket::AudioOptions())));
    auto result_or_error = peer_connection_->AddTrack(audio_track, {kStreamId});
    if (!result_or_error.ok()) {
        RTC_LOG(LS_ERROR) << "Failed to add audio track to PeerConnection: "
                          << result_or_error.error().message();
    }

    rtc::scoped_refptr<ExternalCapturerTrackSource> video_device =
            ExternalCapturerTrackSource::Create();


    if (video_device) {
        rtc::scoped_refptr<webrtc::VideoTrackInterface> video_track_(
                peer_connection_factory_->CreateVideoTrack(kVideoLabel, video_device));
        main_wnd_->StartLocalRenderer(video_track_);

        result_or_error = peer_connection_->AddTrack(video_track_, {kStreamId});
        if (!result_or_error.ok()) {
            RTC_LOG(LS_ERROR) << "Failed to add video track to PeerConnection: "
                              << result_or_error.error().message();
        }
    } else {
        RTC_LOG(LS_ERROR) << "OpenVideoCaptureDevice failed";
    }

    main_wnd_->SwitchToStreamingUI();
}

void Conductor::DisconnectFromCurrentPeer() {
    RTC_LOG(INFO) << __FUNCTION__;
    if (peer_connection_.get()) {
        client_->SendHangUp(peer_id_);
        DeletePeerConnection();
    }

    if (main_wnd_->IsWindow())
        main_wnd_->SwitchToPeerList(client_->peers());
}

void Conductor::UIThreadCallback(int msg_id, void* data) {
    switch (msg_id) {
        case PEER_CONNECTION_CLOSED:
            RTC_LOG(INFO) << "PEER_CONNECTION_CLOSED";
            DeletePeerConnection();

            if (main_wnd_->IsWindow()) {
                if (client_->is_connected()) {
                    main_wnd_->SwitchToPeerList(client_->peers());
                } else {
                    main_wnd_->SwitchToConnectUI();
                }
            } else {
                DisconnectFromServer();
            }
            break;

        case SEND_MESSAGE_TO_PEER: {
            RTC_LOG(INFO) << "SEND_MESSAGE_TO_PEER";
            std::string* msg = reinterpret_cast<std::string*>(data);
            if (msg) {
                // For convenience, we always run the message through the queue.
                // This way we can be sure that messages are sent to the server
                // in the same order they were signaled without much hassle.
                pending_messages_.push_back(msg);
            }

            if (!pending_messages_.empty() && !client_->IsSendingMessage()) {
                msg = pending_messages_.front();
                pending_messages_.pop_front();

                if (!client_->SendToPeer(peer_id_, *msg) && peer_id_ != -1) {
                    RTC_LOG(LS_ERROR) << "SendToPeer failed";
                    DisconnectFromServer();
                }
                delete msg;
            }

            if (!peer_connection_.get())
                peer_id_ = -1;

            break;
        }

        case NEW_TRACK_ADDED: {
            auto* track = reinterpret_cast<webrtc::MediaStreamTrackInterface*>(data);
            if (track->kind() == webrtc::MediaStreamTrackInterface::kVideoKind) {
                auto* video_track = static_cast<webrtc::VideoTrackInterface*>(track);
                main_wnd_->StartRemoteRenderer(video_track);
            }
            track->Release();
            break;
        }

        case TRACK_REMOVED: {
            // Remote peer stopped sending a track.
            auto* track = reinterpret_cast<webrtc::MediaStreamTrackInterface*>(data);
            track->Release();
            break;
        }

        default:
            RTC_NOTREACHED();
            break;
    }
}

void Conductor::OnSuccess(webrtc::SessionDescriptionInterface* desc) {
    peer_connection_->SetLocalDescription(
            DummySetSessionDescriptionObserver::Create(), desc);

    std::string sdp;
    desc->ToString(&sdp);

    // For loopback test. To save some connecting delay.
    if (loopback_) {
        // Replace message type from "offer" to "answer"
        std::unique_ptr<webrtc::SessionDescriptionInterface> session_description =
                webrtc::CreateSessionDescription(webrtc::SdpType::kAnswer, sdp);
        peer_connection_->SetRemoteDescription(
                DummySetSessionDescriptionObserver::Create(),
                session_description.release());
        return;
    }

    json jmessage;
    jmessage[kSessionDescriptionTypeName] =
            webrtc::SdpTypeToString(desc->GetType());
    jmessage[kSessionDescriptionSdpName] = sdp;

    SendMessage(jmessage.dump(8));
}

void Conductor::OnFailure(webrtc::RTCError error) {
    RTC_LOG(LERROR) << ToString(error.type()) << ": " << error.message();
}

void Conductor::SendMessage(const std::string& json_object) {
    std::string* msg = new std::string(json_object);
    UIThreadCallback(SEND_MESSAGE_TO_PEER, msg);
}
